// Copyright 2023 Huawei Cloud Computing Technology Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "RedisHandler.h"
#include <string>
#include <cstring>
#include <iostream>
#include "Log.h"

using namespace std;

RedisHandler::RedisHandler()
{
    m_addr = "";
    m_port = 0;
    m_pwd = "";
    pm_rct = NULL;
    pm_rr = NULL;
    error_msg = "";
}

RedisHandler::~RedisHandler()
{
    disConnect();
    pm_rct = NULL;
    pm_rr = NULL;
}

int RedisHandler::connect(const string &addr = "127.0.0.1", int port = 6379, const string &pwd)
{
    m_addr = addr;
    m_port = port;
    m_pwd = pwd;

    pm_rct = redisConnect(m_addr.c_str(), m_port);

    if (pm_rct->err) {
        error_msg = pm_rct->errstr;
        return M_CONNECT_FAIL;
    }

    if (NULL == pm_rct) {
        return M_CONNECT_FAIL;
    }

    if (!m_pwd.empty()) {
        return connectAuth(m_pwd);
    }

    return M_REDIS_OK;
}

int RedisHandler::disConnect()
{
    redisFree(pm_rct);
    freeReplyObject(pm_rr);
}

void RedisHandler::free()
{
    if(pm_rr != NULL){
        freeReplyObject(pm_rr);
        pm_rr = NULL;
    }
}

int RedisHandler::setValue(const string &key, const string &value)
{
    string cmd = "set " + key + " " + value;
    pm_rr = (redisReply *) redisCommand(pm_rct, cmd.c_str());
    return handleReply();
}

int RedisHandler::getValue(const string &key, string &value)
{
    string cmd = "get " + key;
    pm_rr = (redisReply *) redisCommand(pm_rct, cmd.c_str());
    int ret = handleReply(&value);
    return ret;
}

int RedisHandler::tryLock(const string &key, const string &value, int expireTime)
{
    string cmd = "set " + key + " " + value + " ex " + to_string(expireTime) + " nx";
    pm_rr = (redisReply *) redisCommand(pm_rct, cmd.c_str());
    int ret = handleReply();
    return ret;
}

int RedisHandler::delKey(const string &key)
{
    string cmd = "del " + key;
    pm_rr = (redisReply *) redisCommand(pm_rct, cmd.c_str());
    int rows = 0;
    int ret = handleReply(&rows);
    if (ret == M_REDIS_OK && rows == 0) {
        LOG_WARN_FMT("Del key is not exit");
    }
    return ret;
}

int RedisHandler::findKeys(std::vector<std::string> &keys, string &cmd)
{
    pm_rr = (redisReply *) redisCommand(pm_rct, cmd.c_str());

    int len;
    redisReply **array;
    int ret = handleReply(&len, &array);
    if (ret == M_REDIS_OK) {
        for (int i = 0; i < len; i++) {
            keys.push_back(string(array[i]->str));
        }
        free();
        return ret;
    }
    free();

    return ret;
}

int RedisHandler::findValues(std::vector<std::string> &values, string &cmd)
{
    pm_rr = (redisReply *) redisCommand(pm_rct, cmd.c_str());

    int len;
    redisReply **array;
    int ret = handleReply(&len, &array);
    if (ret == M_REDIS_OK) {
        for (int i = 0; i < len; i++) {
            values.push_back(string(array[i]->str));
        }
        free();
        return ret;
    }
    free();

    return ret;
}

int RedisHandler::findAllKeys(std::vector<std::string> &keys)
{
    string cmd = "keys *";

    pm_rr = (redisReply *) redisCommand(pm_rct, cmd.c_str());

    int len;
    redisReply **array;
    int ret = handleReply(&len, &array);
    LOG_DEBUG_FMT("find all key value return ret = %d", ret);
    if (ret == M_REDIS_OK) {
        for (int i = 0; i < len; i++) {
            keys.push_back(string(array[i]->str));
        }
        free();
        return ret;
    }
    free();
    
    return ret;
}

int RedisHandler::printAll()
{
    string cmd = "keys *";

    pm_rr = (redisReply *) redisCommand(pm_rct, cmd.c_str());

    int len;
    redisReply **array;
    int ret = handleReply(&len, &array);
    if (ret == M_REDIS_OK) {
        for (int i = 0; i < len; i++)
            cout << string(array[i]->str) << endl;
    } else
        return 0;
}

string RedisHandler::getErrorMsg()
{
    return error_msg;
}

int RedisHandler::connectAuth(const string &psw)
{
    string cmd = "auth " + psw;

    pm_rr = (redisReply *) redisCommand(pm_rct, cmd.c_str());

    return handleReply();
}

int RedisHandler::handleReply(void *value, redisReply ***array)
{
    if (pm_rct->err) {
        LOG_INFO_FMT("Pm rct err is not null");
        error_msg = pm_rct->errstr;
        return M_CONTEXT_ERROR;
    }

    if (pm_rr == NULL) {
        error_msg = "auth redisReply is NULL";
        return M_REPLY_ERROR;
    }

    switch (pm_rr->type) {
        case REDIS_REPLY_ERROR:
            error_msg = pm_rr->str;
            return M_EXE_COMMAND_ERROR;
        case REDIS_REPLY_STATUS:
            if (!strcmp(pm_rr->str, "OK"))
                return M_REDIS_OK;
            else {
                error_msg = pm_rr->str;
                return M_EXE_COMMAND_ERROR;
            }
        case REDIS_REPLY_INTEGER:
            *(int *) value = pm_rr->integer;
            return M_REDIS_OK;
        case REDIS_REPLY_STRING:
            *(string *) value = pm_rr->str;
            return M_REDIS_OK;
        case REDIS_REPLY_NIL:
            return M_EXE_COMMAND_ERROR;
        case REDIS_REPLY_ARRAY:
            *(int *) value = pm_rr->elements;
            *array = pm_rr->element;
            return M_REDIS_OK;
        default:
            error_msg = "unknown reply type";
            return M_EXE_COMMAND_ERROR;
    }
}

RedisOperator::RedisOperator()
{
    pRedis = std::make_shared<RedisHandler>();
}

RedisOperator::~RedisOperator()
{ 
}

int RedisOperator::connect(const string &addr, int port, const string &pwd)
{
    pRedis->reConnectCnt++;

    int ret = pRedis->connect(addr, port, pwd);
    if (ret != M_REDIS_OK) {
        LOG_WARN_FMT("Redis connect error getErrorMsg = %s, ret = %d, errno = %d, addr = %s, port = %d",
            pRedis->getErrorMsg().c_str(), ret, errno, addr.c_str(), port);
    }
    pRedis->reConnectCnt = 0;
    this->free();

    return M_REDIS_OK;
}

int RedisOperator::setValue(const string &key, const string &value)
{
    int ret = pRedis->setValue(key, value);
    if (ret != M_REDIS_OK) {
        LOG_WARN_FMT("Redis setValue error getErrorMsg = %s, ret = %d, errno = %d, key = %s",
            pRedis->getErrorMsg().c_str(), ret, errno, key.c_str());
    }
    this->free();
    
    if(ret == M_CONTEXT_ERROR){
        this->connect(rr::gMasterConfig.iRedis.address,rr::gMasterConfig.iRedis.port, rr::gMasterConfig.iRedis.password);
        pRedis->setValue(key, value);
    }
    return ret;
}

int RedisOperator::getValue(const string &key, string &value)
{
    string midValue;
    int ret = pRedis->getValue(key, midValue);

    if (ret != M_REDIS_OK) {
        LOG_DEBUG_FMT("Redis getValue error getErrorMsg = %s, ret = %d, errno = %d, key = %s",
            pRedis->getErrorMsg().c_str(), ret, errno, key.c_str());
    }
    value = midValue;
    this->free();
    return ret;
}

int RedisOperator::tryLock(const string &key, const string &value, int expireTime)
{
    int ret = pRedis->tryLock(key, value, expireTime);
    if (ret != M_REDIS_OK) {
        LOG_WARN_FMT("Redis try lock getErrorMsg = %s, ret = %d, errno = %d, key = %s",
                     pRedis->getErrorMsg().c_str(), ret, errno, key.c_str());
    }
    this->free();
    return ret;
}

int RedisOperator::delKey(const string &key)
{
    int ret = pRedis->delKey(key);
    if (ret != M_REDIS_OK) {
        LOG_WARN_FMT("Redis del key error getErrorMsg = %s, ret = %d, errno = %d, key = %s",
            pRedis->getErrorMsg().c_str(), ret, errno,  key.c_str());
    }
    this->free();
    return ret;
}

int RedisOperator::findKeys(std::vector<std::string> &keys, string &cmd)
{
    int ret = pRedis->findKeys(keys, cmd);

    if (ret != M_REDIS_OK) {
        LOG_WARN_FMT("Find keys failed errorMsg = %s, ret = %d, errno = %d", pRedis->getErrorMsg().c_str(), ret, errno);
    }

    if(ret == M_CONTEXT_ERROR){
        this->connect(rr::gMasterConfig.iRedis.address,rr::gMasterConfig.iRedis.port, rr::gMasterConfig.iRedis.password);
        pRedis->findKeys(keys, cmd);
    }

    return M_REDIS_OK;
}

int RedisOperator::findValues(std::vector<std::string> &values, string &cmd)
{
    int ret = pRedis->findKeys(values, cmd);

    if (ret != M_REDIS_OK) {
        LOG_WARN_FMT("Find values failed errorMsg = %s, ret = %d, errno = %d", pRedis->getErrorMsg().c_str(), ret, errno);
    }

    if(ret == M_CONTEXT_ERROR){
        this->connect(rr::gMasterConfig.iRedis.address,rr::gMasterConfig.iRedis.port, rr::gMasterConfig.iRedis.password);
        pRedis->findKeys(values, cmd);
    }

    return M_REDIS_OK;
}

int RedisOperator::findAllKeys(std::vector<std::string> &keys)
{
    int ret = pRedis->findAllKeys(keys);
        
    if (ret != M_REDIS_OK) {
        LOG_WARN_FMT("Find all keys failed errorMsg = %s, ret = %d, errno = %d", pRedis->getErrorMsg().c_str(), ret, errno);
    }

    if(ret == M_CONTEXT_ERROR){
        this->connect(rr::gMasterConfig.iRedis.address,rr::gMasterConfig.iRedis.port, rr::gMasterConfig.iRedis.password);
        pRedis->findAllKeys(keys);
    }
    
    return M_REDIS_OK;
}

void RedisOperator::free()
{
    pRedis->free();
}